package com.lagou.observer;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CollectionUtils;

import java.io.IOException;
import java.util.List;

public class DeleteRelationObserver extends BaseRegionObserver {

    public static final String FAMILY_FRIENDS = "friends";
    public static final String TABLE_NAME = "chat_user";

    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
        Table chat = e.getEnvironment().getTable(TableName.valueOf(TABLE_NAME));

        // 获取friends
        List<Cell> cells = put.getFamilyCellMap().get(Bytes.toBytes(FAMILY_FRIENDS));
        if (CollectionUtils.isEmpty(cells)) {
            return;
        }
        // 当 chat_user 的 uid1 (rowKey) 向 friends（family） 插入字段 uid2 时，同时向 uid2 (kowKey) 的 friends（family） 插入字段 uid1，值wei uid1
        for (Cell cell : cells) {
            Get get = new Get(CellUtil.cloneQualifier(cell)).addColumn(Bytes.toBytes(FAMILY_FRIENDS), CellUtil.cloneRow(cell));
            if (chat.exists(get)) {
                System.out.println("exists: "+CellUtil.toString(cell,false));
                continue;
            }
            System.out.println("out: "+CellUtil.toString(cell,false));
            Put p = new Put(CellUtil.cloneQualifier(cell));
            p.addColumn(Bytes.toBytes(FAMILY_FRIENDS),CellUtil.cloneRow(cell), CellUtil.cloneRow(cell));
            chat.put(p);
        }
        chat.close();
    }

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        Table chat = e.getEnvironment().getTable(TableName.valueOf(TABLE_NAME));
        List<Cell> cells = delete.getFamilyCellMap().get(Bytes.toBytes(FAMILY_FRIENDS));
        if (CollectionUtils.isEmpty(cells)) {
            return;
        }
        for (Cell cell : cells) {
            byte[] delRowKey = CellUtil.cloneQualifier(cell);
            if (chat.exists(new Get(delRowKey).addColumn(Bytes.toBytes(FAMILY_FRIENDS), CellUtil.cloneRow(cell)))) {
                Delete del = new Delete(delRowKey);
                chat.delete(del);
            }
        }
        chat.close();
    }
}
